package cn.com.cennavi.service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import org.apache.spark.api.java.function.PairFlatMapFunction;

import cn.com.cennavi.bean.AverageSpeedData;
import cn.com.cennavi.bean.DiDiTrajectory;
import cn.com.cennavi.bean.HighPrecisionSingleDirTraffic;
import scala.Tuple2;

/**
 * 单方向路况计算
 * 
 * @ClassName: SingleDirRoadTrafficComputer
 * @Description: TODO（这里用一句话描述这个类的作用）
 * @author bull
 * @date 2016年4月13日 上午11:13:36
 *
 */
public class SingleDirRoadTrafficComputer implements PairFlatMapFunction<Tuple2<String, DiDiTrajectory>, String, String> {
	private static final long serialVersionUID = 1L;

	@Override
	public Iterator<Tuple2<String, String>> call(Tuple2<String, DiDiTrajectory> arg0) throws Exception {
		// TODO Auto-generated method stub
		ArrayList<Tuple2<String, String>> result = null;
		DiDiTrajectory diDiTrajectory = arg0._2;
		String carId = arg0._1();
		// 计算一辆车的路况
		result = getGPSAverageSpeedInfo(diDiTrajectory, carId);
		return result.iterator();
	}

	private ArrayList<Tuple2<String, String>> getGPSAverageSpeedInfo(DiDiTrajectory diDiTrajectory, String carId) {
		// TODO Auto-generated method stub
		ArrayList<Tuple2<String, String>> result = new ArrayList<>();
		String key1 = diDiTrajectory.toShowString();
		String[] gpsDisArr = diDiTrajectory.gpsDisSeq.split("\\|");
		String[] gpsTimeArr = diDiTrajectory.gpsTimeSeq.split("\\|");
		// 用于保存标准gps点的瞬时速度
//		String[] speedArr = diDiTrajectory.speedSeq.split("\\|");
		if (gpsTimeArr.length <= 1 || gpsDisArr.length <= 1 || (gpsDisArr.length != gpsTimeArr.length))
			return result;
		for (int i = 0; i < gpsTimeArr.length - 1; i++) {
			long timeStamp = Long.parseLong(gpsTimeArr[i]);
			int length = (int) Double.parseDouble(gpsDisArr[i]);
			long afterTimeStamp = Long.parseLong(gpsTimeArr[i + 1]);
			int afterLength = (int) Double.parseDouble(gpsDisArr[i + 1]);
//			String speed = speedArr[i];
			double aveSpeed = (double) (Math.abs((afterLength - length) / (double) (afterTimeStamp - timeStamp)) * 3.6);
			if (aveSpeed > 150) {
				continue;
			}
			if (timeStamp - afterTimeStamp >= 0) {
				continue;
			}
			if (afterLength - length >= 0) {
				continue;
			}

			for (int j = afterLength; j < length; j++) {
				String key = key1 + "_" + j;
				StringBuffer vb = new StringBuffer();
				vb.append(aveSpeed);
				vb.append("_");
				vb.append(timeStamp);
				result.add(new Tuple2<String, String>(key, vb.toString()));
				// String value = diDiTrajectory.timeStamp + "_" +
				// aveSpeed+"_"+speed+"_"+carId+":"+timeStamp;
				// result.add(new Tuple2<String, String>(key, value));
			}
		}
		return result;

	}

	public HighPrecisionSingleDirTraffic getTrafficInfo(AverageSpeedData averageSpeedInfo) throws IOException {
		// TODO Auto-generated method stub
		ArrayList<String> dataList = new LaneSpeedSplitCombine().getClassPattern(averageSpeedInfo.speed_seq, Integer.parseInt("10"), Integer.parseInt("1"), Float.parseFloat("0.1"));
		if (averageSpeedInfo.speed_seq == "" || averageSpeedInfo.speed_seq == null) {
			return null;
		}
		// 转换成输出格式并保存
		// 输出格式定义：difpointId,lon|lat,straightAngle,difPointDirClass,trafficDir,bef_Dif_Len_Speed_seq,aft_Dif_Len_Speed_seq,totalLen,befLen,aft_len
		String difPointId = averageSpeedInfo.didiData.difPointId;
		if (averageSpeedInfo.didiData.angle == null) {
			return null;
		}
		String lonlat = averageSpeedInfo.didiData.lonLat;
		int trafficDir = averageSpeedInfo.didiData.dir;
		long timeStamp = averageSpeedInfo.didiData.timeStamp;
		int difPointDis = averageSpeedInfo.didiData.difPointDisToStartLink;
		/** 计算分歧点前后的长度 **/
		StringBuffer length_speed_seq_bef = new StringBuffer();
		StringBuffer length_speed_seq_aft = new StringBuffer();
		int length_bef = 0;
		boolean isBef = false;
		for (String s : dataList) {
			String[] dis_speed = s.split("\\:");
			int length_1 = Integer.parseInt(dis_speed[0].split("\\_")[1]);
			int length_0 = Integer.parseInt(dis_speed[0].split("\\_")[0]);
			int speed = (int) Double.parseDouble(dis_speed[1]);
			int nowlen = length_1;
			if (length_1 <= difPointDis) {
				/** 在分歧点之后 **/
				length_speed_seq_aft.append((length_1 - length_0) + 1 + "_" + speed);
				length_speed_seq_aft.append("|");
			} else {
				/** 判断是否被分歧点打断 **/
				if (nowlen > difPointDis) {
					if (isBef == true) {
						length_speed_seq_bef.append((length_1 - length_0) + 1 + "_" + speed);
						length_speed_seq_bef.append("|");
						length_bef = nowlen - difPointDis;
					} else {
						/** 被分歧点打断,一部分属于分歧点前，一部分属于分歧点后 **/
						int befLen = length_1 - difPointDis;
						int aftLen = (length_1 - length_0) + 1 - befLen;
						length_speed_seq_bef.append(befLen + "_" + speed);
						length_speed_seq_bef.append("|");
						length_bef = nowlen - difPointDis;
						length_speed_seq_aft.append(aftLen + "_" + speed);
						length_speed_seq_aft.append("|");
					}

				}
				isBef = true;
			}
		}
		int total_length = length_bef + difPointDis + 1;
		int bef_length = length_bef;
		int after_length = difPointDis + 1;

		String line = difPointId + "," + lonlat + "," + averageSpeedInfo.didiData.straightAngle + "," + averageSpeedInfo.didiData.difPointDirClass + "," + trafficDir + "," + timeStamp + "," + length_speed_seq_bef + "," + length_speed_seq_aft + "," + total_length + "," + bef_length + "," + after_length;
		if (bef_length == 0)
			return null;

		return HighPrecisionSingleDirTraffic.parseData(line);
	}

	/**
	 * 速度填补
	 */
	public static String getGPSFillSpeedInfo(AverageSpeedData averageSpeedInfo) {
		// TODO Auto-generated method stub
		String[] gpsInfo = averageSpeedInfo.dis_speed_seq.split("\\|");
		StringBuffer dis_speed_sb = new StringBuffer();
		StringBuffer speed_sb = new StringBuffer();
		for (int i = 0; i < gpsInfo.length - 1; i++) {
			String[] dis_speed = gpsInfo[i].split("\\_");
			int length = Integer.parseInt(dis_speed[0]);
			double Speed = Double.parseDouble(dis_speed[1]);
			String[] aft_dis_speed = gpsInfo[i + 1].split("\\_");
			int aft_length = Integer.parseInt(aft_dis_speed[0]);
			double aft_Speed = Double.parseDouble(aft_dis_speed[1]);
			if (i == 0 && length != 0) {
				for (int j = 0; j < length; j++) {
					dis_speed_sb.append(j);
					dis_speed_sb.append("_");
					dis_speed_sb.append(Speed);
					dis_speed_sb.append("|");

					speed_sb.append(Speed);
					speed_sb.append("|");
				}
			} else {

				if (aft_length - length > 1) {
					double averageSpeed = (Speed + aft_Speed) / 2;
					for (int j = length; j < aft_length; j++) {
						dis_speed_sb.append(j);
						dis_speed_sb.append("_");
						dis_speed_sb.append(averageSpeed);
						dis_speed_sb.append("|");

						speed_sb.append(averageSpeed);
						speed_sb.append("|");
					}
				} else {
					dis_speed_sb.append(i);
					dis_speed_sb.append("_");
					dis_speed_sb.append(Speed);
					dis_speed_sb.append("|");

					speed_sb.append(Speed);
					speed_sb.append("|");
				}
				if (i == gpsInfo.length - 2) {
					dis_speed_sb.append(i + 1);
					dis_speed_sb.append("_");
					dis_speed_sb.append(aft_Speed);
					dis_speed_sb.append("|");

					speed_sb.append(aft_Speed);
					speed_sb.append("|");
				}
			}
		}
		averageSpeedInfo.dis_speed_seq = dis_speed_sb.toString();
		averageSpeedInfo.speed_seq = speed_sb.toString();
		return null;
	}

	/**
	 * 计算gps原始点的平均速度
	 */
	public AverageSpeedData getGPSAverageSpeedInfo(Iterable<DiDiTrajectory> didiList) {
		// TODO Auto-generated method stub
		TreeMap<Integer, List<Double>> dis_speedsMap = new TreeMap<>();
		DiDiTrajectory didiMapInfo = null;
		for (DiDiTrajectory didi : didiList) {
			didiMapInfo = didi;
			String[] gpsDisArr = didi.gpsDisSeq.split("\\|");
			String[] gpsTimeArr = didi.gpsTimeSeq.split("\\|");

			for (int i = 0; i < gpsTimeArr.length - 1; i++) {
				long timeStamp = Long.parseLong(gpsTimeArr[i]);
				int length = (int) Double.parseDouble(gpsDisArr[i]);
				long afterTimeStamp = Long.parseLong(gpsTimeArr[i + 1]);
				int afterLength = (int) Double.parseDouble(gpsDisArr[i + 1]);
				double aveSpeed = (double) (Math.abs((afterLength - length) / (double) (afterTimeStamp - timeStamp)) * 3.6);
				if (timeStamp - afterTimeStamp == 0) {
					continue;
				}
				if (afterLength - length >= 0) {
					continue;
				}

				for (int j = afterLength; j <= length; j++) {
					List<Double> speedList1 = dis_speedsMap.get(j);
					if (speedList1 == null) {
						List<Double> speedList = new ArrayList<>();
						speedList.add(aveSpeed);
						dis_speedsMap.put(j, speedList);
					} else {
						speedList1.add(aveSpeed);
					}

				}
			}
		}
		TreeMap<Integer, Double> dis_speedMap = new TreeMap<>();
		StringBuffer dis_speed_sb = new StringBuffer();
		for (Map.Entry<Integer, List<Double>> entry1 : dis_speedsMap.entrySet()) {
			int dis = entry1.getKey();
			List<Double> speedList = entry1.getValue();
			double ave_speed = 0;
			double sumSpeed = 0;
			for (double speed : speedList) {
				sumSpeed += speed;
			}
			ave_speed = sumSpeed / speedList.size();
			dis_speedMap.put(dis, ave_speed);

		}
		for (Map.Entry<Integer, Double> entry2 : dis_speedMap.entrySet()) {
			dis_speed_sb.append(entry2.getKey());
			dis_speed_sb.append("_");
			dis_speed_sb.append(entry2.getValue());
			dis_speed_sb.append("|");
		}
		AverageSpeedData asd = new AverageSpeedData(didiMapInfo, dis_speed_sb.toString());
		return asd;

	}
}
